package com.guchenbo.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.expressions.TimeIntervalUnit;
import static org.apache.flink.table.api.Expressions.*;

public class FlinkSql {
    public static void main(String[] args) {
        String sourceSql = "CREATE TABLE transactions\n"
                        + "(\n"
                        + "    account_id       BIGINT,\n"
                        + "    amount           BIGINT,\n"
                        + "    transaction_time TIMESTAMP(3),\n"
                        + "    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '5' SECOND\n"
                        + ")\n"
                        + "WITH ( 'connector' = 'kafka',\n"
                        + "    'topic' = 'transactions',\n"
                        + "    'properties.bootstrap.servers' = 'kafka:9092',\n"
                        + "    'scan.startup.mode' = 'earliest-offset',\n"
                        + "    'format' = 'csv')";
        String sinkSql="CREATE TABLE spend_report\n"
                        + "(\n"
                        + "    account_id BIGINT,\n"
                        + "    log_ts     TIMESTAMP(3),\n"
                        + "    amount     BIGINT,\n"
                        + "    PRIMARY KEY (account_id, log_ts) NOT ENFORCED\n"
                        + ")\n"
                        + "WITH ( 'connector' = 'jdbc',\n"
                        + "    'url' = 'jdbc:mysql://mysql:3306/demo',\n"
                        + "    'table-name' = 'spend_report',\n"
                        + "    'driver' = 'com.mysql.jdbc.Driver',\n"
                        + "    'username' = 'gcb',\n"
                        + "    'password' = 'guchenbo')";

        System.out.println(sourceSql);
        System.out.println(sinkSql);

                EnvironmentSettings settings = EnvironmentSettings.newInstance().build();
                TableEnvironment tEnv = TableEnvironment.create(settings);

        tEnv.executeSql(sourceSql);
        tEnv.executeSql(sinkSql);
        Table transactions = tEnv.from("transactions");
        report(transactions).executeInsert("spend_report");

//        tEnv.executeSql("select * from spend_report");
    }
    public static Table report(Table transactions) {
        return transactions.select(
                                        $("account_id"),
                                        $("transaction_time").floor(TimeIntervalUnit.SECOND).as("log_ts"),
                                        $("amount"))
                        .groupBy($("account_id"), $("log_ts"))
                        .select(
                                        $("account_id"),
                                        $("log_ts"),
                                        $("amount").sum().as("amount"));
    }
}
